package repeater

import (
    "../common"
    "database/sql"
    "fmt"
    "github.com/lib/pq"
    "strings"
    "time"
)

var TableColumnMapping = map[string][]string{
    "golia_data_cpu":        []string{"ts", "ip", "metric", "value"},
    "golia_data_mem":        []string{"ts", "ip", "metric", "value"},
    "golia_data_netio":      []string{"ts", "ip", "metric", "value"},
    "golia_data_other":      []string{"ts", "ip", "metric", "value"},
    "golia_data_cpu_core":   []string{"ts", "ip", "metric", "value", "path"},
    "golia_data_diskio":     []string{"ts", "ip", "metric", "value", "path"},
    "golia_data_disk_usage": []string{"ts", "ip", "metric", "value", "path"},
}

var MetricHeadsTableMapping = map[string]string{
    "cpu.cpu_times": "golia_data_cpu",
    "mem.percent":   "golia_data_mem",
    "net.iorate":    "golia_data_netio",
    "host":          "golia_data_other",
    "Misc":          "golia_data_other",
    "load":          "golia_data_other",
    "cpu.core":      "golia_data_cpu_core",
    "disk.iorate":   "golia_data_diskio",
    "disk.percent":  "golia_data_disk_usage",
}


func extractDatPack(dpp *DataPointPack) (ret map[string][][]interface{}) {
    ret = make(map[string][][]interface{})
    for k, _ := range TableColumnMapping {
        ret[k] = make([][]interface{}, 0)
    }
    for _, dp := range dpp.Datapoints {
        tbname, _, values := extract(dp)
        if _, ok := TableColumnMapping[tbname]; ok {
            ret[tbname] = append(ret[tbname], values)
        }
    }
    return
}


func extract(dp *common.DataPoint) (tableName string, names, values []interface{}) {
    ret := common.Extract(dp.Name)
    if len(ret) == 0 {
        return
    }
    var ok bool
    var addr, metric_head, subMetric, path string
    if addr, ok = ret["ip"]; !ok {
        return
    }
    addr = strings.Replace(addr, "_", ".", -1)
    if metric_head, ok = ret["head"]; !ok {
        return
    }
    if subMetric, ok = ret["sub"]; !ok {
        return
    }
    if tableName, ok = MetricHeadsTableMapping[metric_head]; !ok {
        return
    }
    path, ok = ret["path"]
    if ok {
        names = []interface{}{tableName, ",path", ",$5"}
        values = []interface{}{time.Unix(int64(dp.Time), 0), addr, subMetric, dp.Val, path}
        //values = []interface{}{dp.Time, addr, subMetric, dp.Val, path}
    } else {
        names = []interface{}{tableName, "", ""}
        values = []interface{}{time.Unix(int64(dp.Time), 0), addr, subMetric, dp.Val}
        //values = []interface{}{dp.Time, addr, subMetric, dp.Val}
    }
    return
}

//func (cs *PostgreWriter) exec(dp *common.DataPointWithOffset) error {
//    tableName, names, values := extract(dp)
//    if tableName == "" || len(names) == 0 || len(values) == 0 {
//        return nil
//    }
//    if _, prepared := cs.stmts[tableName]; !prepared {
//        sql := fmt.Sprintf("INSERT INTO %s (ts,Ip,metric,Value%s) VALUES (to_timestamp($1),$2,$3,$4%s) ", names...)
//        if cs.upsert {
//            sql = sql + fmt.Sprintf("ON CONFLICT (ts,metric,Ip%s) DO UPDATE SET Value =$4", names[2])
//        }
//        if stmt, err := cs.db.Prepare(sql); err != nil {
//            return err
//        } else {
//            cs.stmts[tableName] = stmt
//        }
//    }
//    _, err := cs.stmts[tableName].Exec(values...)
//    return err
//}

type PostgreWriter struct {
    dbURL         string
    name          string
    upsert        bool
    workingOffset int64
    db            *sql.DB
    rc            *RepeaterController
    stmts         map[string]*sql.Stmt
}

func NewPostgreWriter(cfg map[string]string) (*PostgreWriter, error) {
    url, ok := cfg["url"]
    if !ok {
        return nil, fmt.Errorf("config missing: [url]")
    }
    name, ok := cfg["name"]
    if !ok {
        return nil, fmt.Errorf("config missing: [name]")
    }
    upsert := false
    if us, ok := cfg["upsert"]; ok && us == "true" {
        upsert = true
    }
    return &PostgreWriter{
        dbURL:         url,
        name:          name,
        upsert:        upsert,
        workingOffset: DEFAULT_OFFSET,
    }, nil
}
func (cs *PostgreWriter) Init() (err error) {
    cs.db, err = sql.Open("postgres", cs.dbURL)
    cs.stmts = make(map[string]*sql.Stmt)
    return
}
func (cs *PostgreWriter) Start() {
    cs.rc.wg.Add(1)
}
func (cs *PostgreWriter) Stop() error {
    defer cs.rc.waitGroup().Done()
    return cs.db.Close()
}
func (cs *PostgreWriter) Name() string {
    return "pg|" + cs.name
}

func (cs *PostgreWriter) Handle(dps *DataPointPack) error {
    ret := extractDatPack(dps)
    if len(ret) == 0 {
        log.Warning("no valid datapoint in this batch")
        return nil
    }
    txn, err := cs.db.Begin()
    if err != nil {
        return err
    }
    for tbn, values := range ret {
        columns := TableColumnMapping[tbn]
        stmt, err := txn.Prepare(pq.CopyIn(tbn, columns...))
        if err != nil {
            log.Errorf("prepare stmt err: %v", err)
            return err
        }
        for _, v := range values {
            if _, err = stmt.Exec(v...); err != nil {
                log.Errorf("exec stmt err: %v, ,%v", err,v)
                return err
            }
        }
        if _, err = stmt.Exec(); err != nil {
            log.Errorf("exec stmt err: %v", err)
            return err
        }
        if err = stmt.Close(); err != nil {
            log.Errorf("close stmt err: %v", err)
            return err
        }
    }
    if err = txn.Commit(); err != nil {
        log.Errorf("commit stmt err: %v", err)
        return err
    }
    cs.workingOffset = dps.Offset()
    return nil
}

//func (cs *PostgreWriter) Handle(dp *common.DataPointWithOffset) (*common.DataPointWithOffset, error) {
//    var err error
//    err = cs.exec(dp)
//    if err != nil {
//        sqlerr, ok := err.(*pq.Error)
//        if ok && !cs.upsert && sqlerr.Code == "23505" {
//            err = nil
//        }
//    }
//    cs.workingOffset = dp.Offset
//    return dp, err
//}
func (cs *PostgreWriter) SetController(rc *RepeaterController) error {
    cs.rc = rc
    return nil
}
func (cs *PostgreWriter) Commit() (int64, error) {
    return cs.workingOffset, cs.db.Ping()
}
